package com.yun.util.kafka_es.kafkaTest;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;


public class KafkaSendTest{

    private Map<String, Object> props = new HashMap<>();
    private String keystorePath = "D:\\publish-platform\\platform\\server.keystore.jks";
    private String truststorePath = "D:\\publish-platform\\platform\\server.truststore.jks";
    private KafkaProducer<String, String> kafkaProducer;

    private final static KafkaSendTest sendTest = new KafkaSendTest();

    public static KafkaSendTest getInstance(){
        return sendTest;
    }

    public KafkaSendTest() {
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.6:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 81920);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 134217728);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);// 延迟5ms等待消息填满缓冲区
        props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "platform");// TODO
        // 密码
        props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "platform");// TODO 密码
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "platform");// TODO
        // 密码
        props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "JKS");
        props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "JKS");
        props.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS);
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");// 将这个配置项加上,而且value值为空值,即可解决"不存在主题替代名称"这个问题
        props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keystorePath);
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);

        kafkaProducer = new KafkaProducer<>(props);
    }


    public void send(String topic, String message) {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, message);
        Future<RecordMetadata> send = kafkaProducer.send(producerRecord);
        try {
            send.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
